package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo2RpcSubmit {
  def main(args: Array[String]): Unit = {

    /**
      * createRemoteEnvironment: 远程提交任务
      *
      * 将艾玛写好之后打包，再运行代码
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment
      .createRemoteEnvironment(
        "master",
        40150,
        "C:\\Users\\qx\\IdeaProjects\\BigData13\\flink\\target\\flink-1.0.jar"
      )

    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)
    val wordsDS: DataStream[String] = lineDS.flatMap(_.split(","))
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)
    val countDS: DataStream[(String, Int)] = keyByDS.reduce((x, y) => (x._1, x._2 + y._2))

    countDS.print()

    env.execute()

  }

}
